package com.atguigu.gmall.realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.GmallConfig;
import com.atguigu.gmall.realtime.util.DimUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Collection;
import java.util.Set;

/**
 * @author: xu
 * @desc: 将维度数据通过Phoenix向Hbase表中写数据
 */
public class DimSink extends RichSinkFunction<JSONObject> {
    /**
     * 定义Phoenix连接对象
     */
    private Connection conn = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        // 对连接对象进行初始化
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        conn = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
    }

    /**
     * 对流中的数据进行，生成语句提交 hbase
     */
    @Override
    public void invoke(JSONObject jsonObj, Context context) throws Exception {
        // {"database":"gmall","data":{"tm_name":"liangliang","id":13},"commit":true,"sink_table":"dim_base_trademark","type":"insert","table":"base_trademark","ts":1628134550}
        // 获取目标表名和data数据，data数据就是经过过滤之后，保留的业务表中字段
        String tableName = jsonObj.getString("sink_table");
        JSONObject dataJsonObj = jsonObj.getJSONObject("data");

        if (dataJsonObj != null && dataJsonObj.size() > 0) {
            // 根据data中属性名和属性值生成upsert语句
            String upsertSql = genUpsertSql(tableName.toUpperCase(), dataJsonObj);
            System.out.println("向Phoenix插入数据的SQL:" + upsertSql);
            // 执行SQL
            try (PreparedStatement ps = conn.prepareStatement(upsertSql)) {
                ps.executeUpdate();
                // 注意：执行完Phoenix插入操作之后，需要手动提交事务
                conn.commit();
            } catch (SQLException e) {
                e.printStackTrace();
                throw new RuntimeException("向Phoenix插入数据失败");
            }
            // 如果当前维度数据做的是删除或者修改操作，需要将Redis中缓存的数据清除掉
            if (jsonObj.getString("type").equals("update") || jsonObj.getString("type").equals("delete")) {
                DimUtil.deleteCached(tableName, dataJsonObj.getString("id"));
            }
        }
    }

    /**
     * 根据data属性和值生成向Phoenix中插入数据的sql语句
     */
    private String genUpsertSql(String tableName, JSONObject dataJsonObj) {
        // "upsert into 表空间.表名(列名.....) values (值....)"
        Set<String> keySet = dataJsonObj.keySet();
        Collection<Object> values = dataJsonObj.values();

        // keySet.mkString(",");  =>  "id,tm_name"
        return "upsert into " + GmallConfig.HBASE_SCHEMA + "." + tableName + "(" +
                StringUtils.join(keySet, ",") + ") values('" +
                StringUtils.join(values, "','") + "')";
    }
}
